package com.eshipenet.shipnet.equipmentextbuilder

import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction
import org.apache.flink.util.Collector

class EquipmentWithModelActionGenerator(desc: MapStateDescriptor[String, EquipmentModel])
    extends KeyedBroadcastProcessFunction[String,
                                          EquipmentWithModelIdAction,
                                          EquipmentModelAction,
                                          EquipmentWithModelAction] {

  override def processBroadcastElement(value: EquipmentModelAction,
                                       ctx: KeyedBroadcastProcessFunction[String,
                                                                          EquipmentWithModelIdAction,
                                                                          EquipmentModelAction,
                                                                          EquipmentWithModelAction]#Context,
                                       out: Collector[EquipmentWithModelAction]): Unit = {
    val models = ctx.getBroadcastState(desc)
    value match {
      case EquipmentModelInsert(id, after) =>
        models.put(after.id, after)
      case EquipmentModelUpdate(id, before, after) =>
      case EquipmentModelDelete(id, before)        =>
    }
  }

  override def processElement(value: EquipmentWithModelIdAction,
                              ctx: KeyedBroadcastProcessFunction[String,
                                                                 EquipmentWithModelIdAction,
                                                                 EquipmentModelAction,
                                                                 EquipmentWithModelAction]#ReadOnlyContext,
                              out: Collector[EquipmentWithModelAction]): Unit = ???

}
